package com.zyx.flinkdemo.dataset.iteration;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.IterativeDataSet;

/**
 * @author Yaxi.Zhang
 * @since 2021/5/26 17:01
 * desc: 全量迭代案例
 */
public class IteratePi {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        int maxIterations = 100000;
        // 创建初始的IterativeDataSet
        IterativeDataSet<Integer> initial = env.fromElements(0, 0, 0, 0, 0, 0, 0, 0).iterate(maxIterations);
        DataSet<Integer> iteration = initial.map(new MapFunction<Integer, Integer>() {
            @Override
            public Integer map(Integer count) throws Exception {
                double x = Math.random();
                double y = Math.random();
                return count + ((x * x + y * y) < 1 ? 1 : 0);
            }
        });
        // 迭代转换IterativeDataSet
        DataSet<Integer> count = initial.closeWith(iteration);
        count
                .map(new MapFunction<Integer, Double>() {
                    @Override
                    public Double map(Integer count) throws Exception {
                        return count * 1.0 / maxIterations * 4;
                    }
                })
                .print();
    }
}
